package com.whale.mq.kafka;


import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.stereotype.Service;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.nio.charset.StandardCharsets;

/**
 * Kafka消息发送实现类
 */
@Slf4j
@Service
public class KafkaSendMqServiceImpl   {

    private final String topic = "order-topic"; // 要发送消息的主题
    @Autowired
    private KafkaTemplate<String, byte[]> kafkaTemplate;

    public boolean send(String jsonValue) {
        try {
            // 日志记录发送消息
            log.info("发送消息到Kafka，topic：{}，消息内容：{}", topic, jsonValue);

            // 将字符串消息转换为字节数组并发送
            ListenableFuture<SendResult<String, byte[]>> future =
                    kafkaTemplate.send(topic, MessageBuilder.withPayload(jsonValue.getBytes(StandardCharsets.UTF_8)).build().getPayload());
            // 添加发送成功的回调
            future.addCallback(new ListenableFutureCallback<SendResult<String, byte[]>>() {
                @Override
                public void onSuccess(SendResult<String, byte[]> result) {
                    log.info("消息发送成功，主题：{}，消息偏移量：{}", topic, result.getRecordMetadata().offset());
                }
                @Override
                public void onFailure(Throwable ex) {
                    log.error("消息发送失败", ex);
                }
            });
        } catch (Exception e) {
            // 异常处理
            log.error("发送消息到Kafka时发生异常", e);
            return false;
        }
        return true;
    }

}
